package ruc.DolphinDB.read;

import com.xxdb.DBConnection;
import ruc.Consts.constant;
import ruc.DolphinDB.DolphinDB_Common;
import ruc.utils.IOUtils;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;

public class DolphinDB_ConcurrentRead {
    private int bound, port;
    private String IP, tbName;

    public DolphinDB_ConcurrentRead(String IP, int port, String tbName,int bound) {
        this.IP = IP;
        this.port = port;
        this.tbName = tbName;
        this.bound = bound;
    }

    private String getQuery(int offset, int limit) {
        String format = "select * from %s limit %d, %d";
        return String.format(format, tbName, offset, limit);
    }
    // 1
    // 2
    private void concurrentRead(int client[]) throws InterruptedException, IOException {
        Random ran = new Random();
        DBConnection con = DolphinDB_Common.getDBConnection(IP, port);
        for(int i = 0; i < client.length; i++) {
            CountDownLatch CDL = new CountDownLatch(client[i]);
            AtomicLong AL = new AtomicLong(0);

            for(int k = 0; k < client[i]; k++) {
                long st = System.nanoTime();
                int limit = 0;
                while(limit  == 0) { // 保证limit不为0
                    limit = ran.nextInt(bound);
                }

                con.run(getQuery(ran.nextInt(bound), limit));
                AL.addAndGet((System.nanoTime() - st) / constant.Ns2Ms);
                CDL.countDown();
            }

            CDL.await();
            System.out.printf("[TSBench-D] >>>>> [DolphinDB_ConcurrentRead] ThreadNum:%d, average response time : %d ms>>>>> \n", client[i], AL.longValue() / (long)client[i]);
            IOUtils.WriteData(constant.GetWriteFilePath("DolphinDB", "concurrentRead"),
                    new String[]{String.valueOf(System.nanoTime()), String.valueOf(client[i]), String.valueOf(AL.longValue() / (long)client[i])});
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        DolphinDB_ConcurrentRead db = new DolphinDB_ConcurrentRead(constant.IP, constant.DolphinDB_Port, constant.Concurrent_tbName, constant.bound);
        db.concurrentRead(constant.ConcurrentReadThreadNum);
    }
}
